package cn.com.greatwall.kafka.util;

import java.io.Closeable;
import java.io.IOException;
import java.util.List;

import org.apache.http.HttpHost;
import org.apache.http.auth.AuthScope;
import org.apache.http.auth.UsernamePasswordCredentials;
import org.apache.http.client.CredentialsProvider;
import org.apache.http.impl.client.BasicCredentialsProvider;
import org.apache.http.impl.nio.client.HttpAsyncClientBuilder;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;
import org.elasticsearch.client.RestHighLevelClient;

import com.alibaba.fastjson.JSONObject;

import cn.com.greatwall.common.utils.StringUtils;

/**
 * @Author herw
 * @Time 2021-04-16 14:23:31
 * @Version 1.0
 * @Description: TODO(用一句话描述该文件做什么)
 */
public class ElasticsearchUtil implements Closeable {
    private RestHighLevelClient client = null;
    
    public ElasticsearchUtil(String hosts) {
        this(hosts, "", "");
    }
    
    public ElasticsearchUtil(String hosts, String user, String password) {
        if(!"".equalsIgnoreCase(hosts) && null != hosts) {
            String[] addressArray = hosts.split(",", -1);
            HttpHost[] httpHosts = new HttpHost[addressArray.length];
            
            for (int i = 0; i < addressArray.length; i++) {
                String[] hostAndPort = addressArray[i].split(":", -1);
                httpHosts[i] = new HttpHost(hostAndPort[0], Integer.parseInt(hostAndPort[1]), "http");
            }
            if(StringUtils.isNotBlank(user) && StringUtils.isNotBlank(password)) {
                final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
                credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials(user, password));
                client = new RestHighLevelClient(RestClient.builder(httpHosts).setHttpClientConfigCallback(new RestClientBuilder.HttpClientConfigCallback() {
                    @Override
                    public HttpAsyncClientBuilder customizeHttpClient(HttpAsyncClientBuilder httpClientBuilder) {
                        httpClientBuilder.disableAuthCaching();
                        return httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider);
                    }
                }));
            } else {
                client = new RestHighLevelClient(RestClient.builder(httpHosts));
            }
        }
    }
    
    public boolean batchInsertES(List<JSONObject> datas, String index) throws IOException {
        if(client == null) {
            return false;
        }
        if(datas == null || datas.isEmpty()) {
            return true;
        }
        BulkRequest request = new BulkRequest();
        datas.forEach(data -> {
            request.add(new IndexRequest(index).source(data));
        });
        BulkResponse bulkResponse = client.bulk(request, RequestOptions.DEFAULT);
        return !bulkResponse.hasFailures();
    }

    @Override
    public void close() throws IOException {
        if (client != null) {
            client.close();
        }
    }
}
